package mq.rabbit.demo;

import com.rabbitmq.client.*;
import mq.rabbit.common.util.RabbitMQUtils;
import org.assertj.core.util.Lists;
import org.junit.Test;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Routing之订阅模型-Topic
 * <p>
 * <br />
 * Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。
 * 只不过Topic类型Exchange可以让队列在绑定RoutingKey的时候使用通配符。
 * <p>
 * 这种类型RoutingKey一般都是由一个或多个单词组成，多个单词之间以“.”分隔，例如：item.insert
 * <p>
 * # 通配符：
 * <li>1. * (star) can substitute for exactly one word.   匹配一个词</li>
 * <li>2. # (hash) can substitute for zero or more words. 匹配一个或多个词</li>
 * <p>
 * # 例如：
 * <li>audit.*  只能匹配audit.irs</li>
 * <li>audit.#  可以匹配audit.irs.corporate 或者 audit.irs 等</li>
 */
public class Routing_Topic {

    @Test
    public void testProvider() throws IOException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConnection();

        // 获取通道
        Channel channel = connection.createChannel();

        // 声明交换机以及交换机类型
        channel.exchangeDeclare("topics", "topic");

        // 发布消息
//        String routerKey = "user.save";
        String routerKey = "user.save.findAll";
        String message = "这里是topic动态路由模型，routerKey: [" + routerKey + "] 发送的消息";
        channel.basicPublish("topics", routerKey, null, message.getBytes());

        // 关闭
        RabbitMQUtils.closeChanelAndConnection(channel, connection);
    }

    @Test
    public void testComsuer() throws IOException, InterruptedException {

        /*
         *
         * 使用“*”匹配，只能匹配一个词
         *
         * 使用“#”匹配，可以匹配一个或多个词
         */
        List<String> routingKeys = Lists.newArrayList("user.#", "user.*");

        for (String routingKey : routingKeys) {


            // 获取连接
            Connection connection = RabbitMQUtils.getConnection();
            Channel channel = connection.createChannel();

            // 声明交换机以及交换机类型
            channel.exchangeDeclare("topics", "topic");

            // 创建一个临时队列
            String queueName = channel.queueDeclare().getQueue();

            // 绑定队列和交换机，基于动态通配符形式
            // # 匹配一个或多个词
            channel.queueBind(queueName, "topics", routingKey);

            // 消费消息
            channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者" + queueName + " >>>> " + new String(body));
                }
            });

        }
        TimeUnit.MINUTES.sleep(3);
    }
}
